package com.david.rocketmq;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @description: 消息队列消费者
 * @create: 2021-05-03-18-53
 * @author: David
 **/
@Getter
@Setter
@ToString
@Configuration
@Slf4j
public class Customer {

    private Integer consumeThreadMin=5;
    private Integer consumeThreadMax=32;
    private Integer consumeMessageBatchMaxSize=1;

    private String groupName="miaosha";

    private String namesrvAddr = "localhost:9876";

    @Resource
    private ConsumeMsgListener consumeMsgListener;

    @PostConstruct
    @Bean
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        log.info("defaultConsumer 正在创建---------------------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        // 设置监听
        consumer.registerMessageListener(consumeMsgListener);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
            consumer.subscribe("HelloTopic", "*");
            consumer.start();
            log.info("consumer 创建成功 groupName={}, topics=HelloTopic, namesrvAddr={}",groupName,namesrvAddr);
        } catch (MQClientException e) {
            log.error("consumer 创建失败!");
        }
        return consumer;
    }
}
